home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2008 February / PCWFEB08.iso / Software / Freeware / Miro 1.0 / Miro_Installer.exe / xulrunner / python / test / schedulertest.py < prev    next >
Encoding:
Python Source  |  2007-11-12  |  2.2 KB  |  61 lines
  1. import unittest
  2. from time import time, sleep
  3. import threading
  4.  
  5. import eventloop
  6. from test.framework import EventLoopTest
  7.  
  8. class SchedulerTest(EventLoopTest):
  9.     def setUp(self):
  10.         self.gotArgs = []
  11.         self.gotKwargs = []
  12.         EventLoopTest.setUp(self)
  13.     
  14.     def callback(self, *args, **kwargs):
  15.         self.gotArgs.append(args)
  16.         self.gotKwargs.append(kwargs)
  17.         if 'stop' in kwargs.keys():
  18.             eventloop.quit()
  19.  
  20.     def testCallbacks(self):
  21.         eventloop.addIdle(self.callback, "foo")
  22.         eventloop.addTimeout(0.1, self.callback, "foo", args=("chris",), 
  23.                 kwargs={'hula':"hula"})
  24.         eventloop.addTimeout(0.2, self.callback, "foo", args=("ben",), 
  25.                 kwargs={'hula':'moreHula', 'stop':1})
  26.         self.runEventLoop()
  27.         self.assertEquals(self.gotArgs[0], ())
  28.         self.assertEquals(self.gotArgs[1], ("chris",))
  29.         self.assertEquals(self.gotArgs[2], ("ben",))
  30.         self.assertEquals(self.gotKwargs[0], {})
  31.         self.assertEquals(self.gotKwargs[1], {'hula':'hula'})
  32.         self.assertEquals(self.gotKwargs[2], {'hula':'moreHula', 'stop':1})
  33.  
  34.     def testQuitWithStuffStillScheduled(self):
  35.         eventloop.addTimeout(0.1, self.callback, "foo", kwargs={'stop':1})
  36.         eventloop.addTimeout(2, self.callback, "foo")
  37.         self.runEventLoop()
  38.         self.assertEquals(len(self.gotArgs), 1)
  39.  
  40.     def testTiming(self):
  41.         startTime = time()
  42.         eventloop.addTimeout(0.2, self.callback, "foo", kwargs={'stop':1})
  43.         self.runEventLoop()
  44.         endTime = time()
  45.         self.assertAlmostEqual(startTime + 0.2, endTime, places=1)
  46.  
  47.     def testLotsOfThreads(self):
  48.         timeouts = [0, 0, 0.1, 0.2, 0.3]
  49.         threadCount = 8
  50.         def thread():
  51.             sleep(0.5)
  52.             for timeout in timeouts:
  53.                 eventloop.addTimeout(timeout, self.callback, "foo")
  54.         for i in range(threadCount):
  55.             t = threading.Thread(target=thread)
  56.             t.start()
  57.         eventloop.addTimeout(1, self.callback, "foo", kwargs={'stop':1})
  58.         self.runEventLoop()
  59.         totalCalls = len(timeouts) * threadCount + 1
  60.         self.assertEquals(len(self.gotArgs), totalCalls)
  61.